package com.atguigu.app

import com.alibaba.fastjson.JSON
import com.atguigu.bean.{CouponAlertInfo, EventInfo}
import com.atguigu.common.logger.GmallConstants
import com.atguigu.utils.{EsUtils, MykafkaUtil}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * saprk es预警
 * 需求：同一设备，5分钟内三次及以上用不同账号登录并领取优惠劵，并且在登录到领劵过程中没有浏览商品。达到以上要求则产生一条预警日志。
 * 同一设备，每分钟只记录一次预警。
 *
 * @author WangJX
 * @date 2019/12/2 16:59 
 * @version 1.0
 */
object AlertApp {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("AlertApp")

    val ssc = new StreamingContext(conf, Seconds(5))

    //从kafka中获取事件数据
    val kafkaDstream = MykafkaUtil.getKafkaStream(GmallConstants.KAFKA_TOPIC_EVENT, ssc)
      .map(_.value())
      .map(jsonString => {
        val info: EventInfo = JSON.parseObject(jsonString, classOf[EventInfo])
        (info.mid, info)
      })


    //同一设备，5分钟内三次及以上用不同账号登录并领取优惠劵,并且在登录到领劵过程中没有浏览商品
    val filterDstram = kafkaDstream
      .window(Seconds(300), Seconds(5))
      .filter {
        case (mid, value) => {
          //判断是否有领取优惠券的点击事件，有则保留数据
          var flag = false
          if (value.evid.equalsIgnoreCase("coupon")) {
            flag = true
          }
          flag
        }
      }
      //判断在登录到领劵过程中没有浏览商品
      .filter {
        case (mid, value) => {
          var flag = false
          if (!value.evid.equalsIgnoreCase("clickItem")) {
            flag = true
          }
          flag
        }
      }
      .groupByKey()
      //对同一mid登录的账号进行聚合、并去重
      .map(rdd => {
        val set: Set[EventInfo] = rdd._2.toSet
        set
      })
      .map {
        rdd => {

          val uids = new java.util.HashSet[String]()
          val itemIds = new java.util.HashSet[String]()
          val events = new java.util.ArrayList[String]

          //          val list = ListBuffer[CouponAlertInfo]()
          for (elem <- rdd) {
            //            val str: String = JSON.toJSONString(elem, new SerializeConfig(true))
            uids.add(elem.uid)
            itemIds.add(elem.itemid)
            events.add(elem.evid)
          }

          //完善事件并设定ES的id
          val ts: Long = System.currentTimeMillis()
          val key  = ts / 1000L / 60L

          (key.toString, CouponAlertInfo(rdd.last.mid, uids, itemIds, events, ts)
          )
        }
      }
      //判断是同一mid登录不同账号是否大于等于3次
      .filter(_._2.uids.size() >= 3)
    /*      .foreachRDD{
            rdd => {
              rdd.foreach{
                info => {
                  val str: String = JSON.toJSONString(info, new SerializeConfig(true))
                  println(str)
                }
              }
            }
          }*/

    //同一设备，每分钟只记录一次预警。(使用ES的PUT的幂等性特性)
    filterDstram.foreachRDD {
      rdd => {
        rdd.foreachPartition {
          info => {
//            val str: String = JSON.toJSONString(info, new SerializeConfig(true))
            EsUtils.insertBulk(GmallConstants.ES_INDEX_ALERT, info.toList)
//            println("spark打印输出：" + info.mkString(",").foreach(println))
          }
        }
      }
    }


    ssc.start()

    ssc.awaitTermination()

  }
}



















